package _08_blocking_queue.my_implements;

import _00_my.exception.TechnicalLimitationException;
import _04_queue.AbstractQueue;
import _08_blocking_queue.BlockingQueue;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author: mornd
 * @dateTime: 2023/6/10 - 2:10
 * <p>
 * 双锁实现
 * 使用两把锁保证效率，put 和 take 各使用一把锁，而不是单锁那样 put 时，所有的 take 操作都得阻塞等待，
 * 减小了锁粒度，这里是待调用 signal()方法时才会获取其对应的锁阻塞，主要的增删操作还是异步执行的
 * 最重要的部分是使用局部变量c来让阻塞的线程级联通知，目前能力有限，能理解但表达不出，参考 LinkedBlockingQueue 类
 */
public class DoubleLockArrayBlockingQueue<E>
        extends AbstractQueue<E> implements BlockingQueue<E> {

    // 存储元素 E 的容器
    final E[] items;
    // 队列已用大小
    final AtomicInteger size = new AtomicInteger();
    // 记录入队元素的下标
    int putIndex;
    // 记录出队元素的下标
    int takeIndex;

    /**
     * 入队锁
     */
    final ReentrantLock putLock = new ReentrantLock();
    // 队列满的条件变量
    private final Condition notFull = putLock.newCondition();

    /**
     * 出队锁
     */
    final ReentrantLock takeLock = new ReentrantLock();
    // 队列空的条件变量
    private final Condition notEmpty = takeLock.newCondition();

    /**
     * 初始化队列
     *
     * @param c 队列最大容量
     */
    public DoubleLockArrayBlockingQueue(int c) {
        if (c <= 0) throw new IllegalArgumentException();
        items = (E[]) new Object[c];
    }

    /**
     * 链接队列末尾的节点
     *
     * @param e 节点
     */
    private void enqueue(E e) {
        items[putIndex] = e;
        if (++putIndex == items.length) {
            putIndex = 0;
        }
    }

    /**
     * 从队列头部移除节点
     *
     * @return 返回的节点
     */
    private E dequeue() {
        E e = items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length) {
            takeIndex = 0;
        }
        return e;
    }

    private void signalNotEmpty() {
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

    private void signalNotFull() {
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

    /**
     * 入队元素，如果队列已满则一直阻塞直到队列不是满的
     */
    public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);
        int c; // 记录入队前的队列大小
        putLock.lockInterruptibly();
        try {
            // 1、检查是否已满，如果已满则等待
            while (isFull()) {
                notFull.await();
            }
            // 2、添加元素
            enqueue(e);
            c = size.getAndIncrement();
            if (c < items.length - 1) {
                // 如果此时队列还未满，就继续唤醒 put 线程
                notFull.signal();
            }
        } finally {
            putLock.unlock();
        }

        /*
            put1 put2 put3
            c 表示入队前的队列大小，如果 c==0 则表示这次 put 是在空队列的基础上入的队
            此时要通知 take 线程队列可以取值了，如果 c!=0 则不用通知 take
            因为第一次调用 take 方法会将自己所有 take 操作的线程释放开
         */
        if (c == 0) {
            // 3、解锁队列为空时poll的阻塞，加在 try 块内会导致死锁
            signalNotEmpty();
        }
    }


    /**
     * 出队，如果队列为空，则一直阻塞直到队列有了一个元素
     */
    public E take() throws InterruptedException {
        E e;
        int c; // 出队前队列的大小
        takeLock.lockInterruptibly();
        try {
            // 1、判断队列是否为空，如果为空则阻塞
            while (isEmpty()) {
                notEmpty.await();
            }

            // 2、出列
            e = dequeue();
            c = size.getAndDecrement();
            /*
               take1 take2 take3
               c>1 表示队列还有元素，会执行下面的代码释放另一个 take 线程，
               如果另一个 take 线程发现 c 还是 >0 则继续释放，直到 c<=1 队列为空为止
               可以保证队列有多少个元素，就可以唤醒多少个 take 被 await()的线程，
               如果 size = 3，其中 take 中也可以只有一个或者两个阻塞线程，没有await()的线程也可以调用signal()不报错
             */
            if (c > 1) {
                // 释放另一个 take await()的线程
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }

        // 3、解除put时队列为满的阻塞
        if (c == items.length) {
            // 从满->不满时去级联唤醒 put 线程
            signalNotFull();
        }
        return e;
    }

    /**
     * 入队，如果是满的，直接返回false
     */
    public boolean offer(E e) {
        Objects.requireNonNull(e);
        int c;
        putLock.lock();
        try {
            if (isFull()) {
                return false;
            }
            enqueue(e);
            c = size.getAndIncrement();
            if (c < items.length - 1) {
                notFull.signal();
            }
        } finally {
            putLock.unlock();
        }

        if (c == 0) {
            signalNotEmpty();
        }
        return true;
    }

    /**
     * 出列，如果队列为空，直接返回false
     */
    public E poll() {
        E e;
        int c;
        takeLock.lock();
        try {
            if (isEmpty()) {
                return null;
            }
            e = dequeue();
            c = size.getAndDecrement();
            if (c > 1) {
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }

        if (c == items.length) {
            signalNotFull();
        }
        return e;
    }

    @Override
    public E peek() {
        if (isEmpty()) {
            return null;
        }
        /*
            只上 takeLock 锁是因为代码能走到这里就说明队列不为空
            队列至少有一个元素，而我只要第一个
            putLock不管怎么操作我第一个元素总是不变的
         */
        takeLock.lock();
        try {
            return isEmpty() ? null : items[takeIndex];
        } finally {
            takeLock.unlock();
        }
    }

    /**
     * 入队，超过指定时间还是满的话，退出等待，返回false
     */
    public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
        Objects.requireNonNull(e);
        int c;
        long nanos = unit.toNanos(timeout);
        putLock.lockInterruptibly();
        try {
            while (isFull()) {
                if (nanos < 0L) {
                    return false;
                }
                nanos = notFull.awaitNanos(nanos);
            }
            enqueue(e);
            c = size.getAndIncrement();
            if (c < items.length - 1) {
                notFull.signal();
            }
        } finally {
            putLock.unlock();
        }

        if (c == 0) {
            signalNotEmpty();
        }
        return true;
    }

    /**
     * 出列，如果队列为空，则等待指定的时间，如果时间已过队列还是为空则返回null
     */
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        E e;
        int c;
        long nanos = unit.toNanos(timeout);
        takeLock.lockInterruptibly();
        try {
            while (isEmpty()) {
                if (nanos <= 0L) {
                    return null;
                }
                nanos = notEmpty.awaitNanos(nanos);
            }
            e = dequeue();
            c = size.getAndDecrement();
            if (c > 1) {
                notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }

        if (c == items.length) {
            signalNotFull();
        }
        return e;
    }


    @Override
    public boolean isEmpty() {
        return size() == 0;
    }

    @Override
    public boolean isFull() {
        return size() == items.length;
    }

    @Override
    public void clear() {
        super.clear();
    }

    @Override
    public int size() {
        return size.get();
    }

    @Override
    public String toString() {
        return Arrays.toString(items);
    }

    @Override
    public Iterator<E> iterator() {
        throw new TechnicalLimitationException();
    }
}
